package cc.rengu.oltp.service.base;

import cc.rengu.jradp.application.BulletinBoard;
import cc.rengu.jradp.application.ServiceEntity;
import cc.rengu.jradp.mods.message.MessageTree;
import cc.rengu.jradp.mods.xml.XmlTree;
import cc.rengu.jradp.service.communication.ServerImpl;
import cc.rengu.jradp.service.communication.impl.HttpServer;
import cc.rengu.jradp.service.communication.impl.HttpsServer;
import cc.rengu.jradp.service.communication.impl.WebSocketSSLServer;
import cc.rengu.jradp.service.communication.impl.WebSocketServer;
import cc.rengu.oltp.utility.util.RedisUtil;
import cc.rengu.utility.base.StringUtil;
import cc.rengu.utility.base.XmlConfigUtil;
import cc.rengu.utility.communication.NoResponseException;
import cc.rengu.utility.communication2.AbstractRServer;
import cc.rengu.utility.communication2.MiniServerProcessImpl;
import cc.rengu.utility.communication2.netty.websocket.server.RWebSocketHandler;
import cc.rengu.utility.log.Level;
import cc.rengu.utility.log.RgLog;
import cc.rengu.utility.log.RgLogger;
import cc.rengu.utility.threadarea.ThreadVar;
import cc.rengu.utility.xml.XmlError;
import cc.rengu.utility.xml.XmlValue;
import com.alibaba.fastjson.JSON;
import redis.clients.jedis.JedisPubSub;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息订阅发布服务
 * Created by wangc on 2020/7/23.
 */
public class MsgPubSubService extends RadpService {
    private AbstractRServer<?, ?> subscribeserver = null;
    private AbstractRServer<?, ?> publishserver = null;
    private final ConcurrentHashMap<String, List<Object>> channelContextMapById = new ConcurrentHashMap<>(10000);
    private final ConcurrentHashMap<Object, String> channelContextMapByCtx = new ConcurrentHashMap<>(10000);
    private final ConcurrentHashMap<Object, Long> channelContextMapTimeout = new ConcurrentHashMap<>(10000);
    private final Set<String> cacheUserIds = Collections.synchronizedSet(new HashSet<>());
    private String serviceName;
    private Level logLevel;
    private long linkDetectTime = 60000;
    private String linkDetectRspMsg = null;
    private boolean cacheMsgFlag = false;
    private int cacheMsgTime = 1800000;

    @Override
    protected int bizInit(String serviceName) {
        this.serviceName = serviceName;
        CommunicationSubscribeProcess subscribeProcess = new CommunicationSubscribeProcess();
        CommunicationPublishProcess publishProcess = new CommunicationPublishProcess();
        subscribeserver = initServiceServer(serviceName, subscribeProcess, "Subscribe");
        publishserver = initServiceServer(serviceName, publishProcess, "Publish");
        new DeameonThread().start();
        new SubscribeMsgPushThread().start();
        return 0;
    }

    @Override
    protected int bizProcess(String serviceName, int i) {
        return 0;
    }

    @Override
    protected int bizTerminate(String serviceName) {
        return 0;
    }

    private AbstractRServer<?, ?> initServiceServer(String name, CommunicationServerProcess serverprocess, String ProName) {
        RgLogger rglog = RgLog.getLogger(this.getClass().getName());
        ServiceEntity ServiceConfig = BulletinBoard.getService(name);
        logLevel = ServiceConfig.getLogLevel();
        XmlConfigUtil configfile = new XmlConfigUtil(ServiceConfig.getConfigFile());
        int poolsize = configfile.getIntResolvedPlaceholders("ThreadPoolSize");
        int port = configfile.getIntResolvedPlaceholders(ProName + "/Port");
        if ("Subscribe".equals(ProName)) {
            String linkDetectTime = configfile.getStringResolvedPlaceholders(ProName + "/LinkDetectTime");
            if (null != linkDetectTime) {
                this.linkDetectTime = Long.parseLong(linkDetectTime);
            }
            String linkDetectRspMsg = configfile.getStringResolvedPlaceholders(ProName + "/LinkDetectRspMsg");
            if (null != linkDetectRspMsg && !linkDetectRspMsg.isEmpty()) {
                this.linkDetectRspMsg = linkDetectRspMsg;
            }
        }
        if ("Publish".equals(ProName)) {
            String cacheMsgFlag = configfile.getStringResolvedPlaceholders(ProName + "/CacheMsgFalg");
            if (cacheMsgFlag != null && cacheMsgFlag.equalsIgnoreCase("yes")) {
                this.cacheMsgFlag = true;
            }
            String cacheMsgTime = configfile.getStringResolvedPlaceholders(ProName + "/CacheMsgTime");
            if (cacheMsgTime != null) {
                this.cacheMsgTime = Integer.parseInt(cacheMsgTime);
            }
        }
        String ServerType = configfile.getStringResolvedPlaceholders(ProName + "/Server/@type");
        ServerImpl serverimpl;
        if (ServerType == null) {
            ServerType = "http";
        }
        switch (ServerType.toLowerCase()) {
            case "http":
            default:
                serverimpl = new HttpServer();
                break;
            case "https":
                serverimpl = new HttpsServer();
                break;
            case "websocket":
                serverimpl = new WebSocketServer();
                break;
            case "websocketSSL":
                serverimpl = new WebSocketSSLServer();
                break;
        }
        AbstractRServer<?, ?> server =
                serverimpl.initServer(name, port, poolsize, serverprocess, null, configfile,
                        ProName + "/Server/", rglog);
        server.init();
        return server;
    }

    private abstract static class CommunicationServerProcess implements MiniServerProcessImpl {
        @Override
        public abstract byte[] proc(byte[] buffer, String url, Object obj, int label) throws IOException;

        @Override
        public String getXmlValue(String arg0) {
            MessageTree messagetree = new MessageTree();
            XmlError error = new XmlError();
            XmlValue value = messagetree.getValue(arg0, error);
            if (value == null) {
                return null;
            } else {
                return value.getText();
            }
        }

        @Override
        public void setXmlValue(String arg0, String arg1) {
            MessageTree messagetree = new MessageTree();
            XmlValue value = new XmlValue(arg1);
            messagetree.setValue(arg0, value);
        }

        @Override
        public void initXml() {
            new XmlTree("radp");
        }

        @Override
        public void termXml() {
            ThreadVar.clear();
        }
    }

    private class CommunicationPublishProcess extends CommunicationServerProcess {
        @Override
        public byte[] proc(byte[] buffer, String url, Object obj, int label) {
            RgLogger rglog = BulletinBoard.getLogger(this.getClass().getName());
            try {
                rglog.setLogLevel(logLevel);
                rglog.createLogKey();
                MessageTree tree = new MessageTree();
                XmlValue value = tree.getValue("_MESSAGEHEAD/http/websocket-userid", null);
                if (value == null) {
                    value = tree.getValue("_MESSAGEHEAD/http/NoticeUserId", null);
                    if (value == null) {
                        rglog.error("没有设置_MESSAGEHEAD/http/NoticeUserId的值");
                        setXmlValue("_MESSAGEHEAD/@server_ret", "500");
                        return new byte[0];
                    }
                }
                String bufStr = StringUtil.binaryToBase64(buffer);
                if (cacheMsgFlag) {
                    String pushkey = "noticeWebSocket:list:" + value.getText();
                    Long listMsgElementNum = RedisUtil.onceRightPush(pushkey, (System.currentTimeMillis() + cacheMsgTime) + "|" + bufStr);
                    cacheUserIds.add(value.getText());
                    rglog.info("共有{}条消息推送给订阅者<{}>，当前要推送消息:{}", listMsgElementNum, value.getText(), JSON.toJSONString(new String(buffer), false));
                } else {
                    rglog.info("成功向订阅者<{}>推送消息:{}", value.getText(), JSON.toJSONString(new String(buffer), false));
                }
                setXmlValue("_MESSAGEHEAD/@server_ret", "200");
            } catch (Exception e) {
                // 最终异常处理逻辑
                rglog.error("{}", StringUtil.ExceptionToString(e));
                setXmlValue("_MESSAGEHEAD/@server_ret", "500");
            }
            return new byte[0];
        }
    }

    private class CommunicationSubscribeProcess extends CommunicationServerProcess {
        @Override
        public byte[] proc(byte[] buffer, String url, Object obj, int label) {
            RgLogger rglog = BulletinBoard.getLogger(this.getClass().getName());
            try {
                rglog.setLogLevel(logLevel);
                rglog.createLogKey();
                switch (label) {
                    case RWebSocketHandler.WEBSOCKET_MSG:
                        if (buffer == null || new String(buffer).isEmpty()) {
                            break;
                        }
                        String userId = new String(buffer);
                        int index = userId.toLowerCase().indexOf("websocket-userid");
                        if (userId.startsWith("{") && userId.endsWith("}") && index > 0) {
                            index = userId.indexOf("\"", index + 17);
                            int indexend = userId.indexOf("\"", index + 1);
                            userId = userId.substring(index + 1, indexend);
                        } else {
                            rglog.error("接收报文[{}]中没有找到订阅者id", JSON.toJSONString(userId, false));
                            break;
                        }
                        List<Object> list = channelContextMapById.get(userId);
                        if (list == null) {
                            list = Collections.synchronizedList(new ArrayList<>());
                            channelContextMapById.put(userId, list);
                            rglog.info("订阅者<{}>首次订阅消息，消息推送通道:{}", userId, obj);
                            list.add(obj);
                        } else if (!list.contains(obj)) {
                            rglog.info("将订阅者<{}>的推送通道[{}]添加到订阅链", userId, obj);
                            list.add(obj);
                        }
                        channelContextMapTimeout.put(obj, System.currentTimeMillis() + linkDetectTime);
                        channelContextMapByCtx.put(obj, userId);
                        if (null == linkDetectRspMsg) {
                            subscribeserver.response(obj, buffer, 0);
                        } else {
                            subscribeserver.response(obj, linkDetectRspMsg.getBytes(), 0);
                        }
                        break;
                    case RWebSocketHandler.WEBSOCKET_CLOSE:
                        userId = channelContextMapByCtx.get(obj);
                        if (userId != null) {
                            rglog.info("订阅者<{}>的推送通道[{}]链路关闭", userId, obj);
                            channelContextMapByCtx.remove(obj);
                            channelContextMapTimeout.remove(obj);
                            list = channelContextMapById.get(userId);
                            if (list != null) {
                                list.remove(obj);
                                rglog.info("从订阅者<{}>的订阅链中删除推送通道[{}],还剩{}个推送通道", userId, obj, list.size());
                            }
                        } else {
                            rglog.error("没有找到用户的推送通道");
                        }
                        break;
                    case RWebSocketHandler.WEBSOCKET_PING:
                        if (channelContextMapTimeout.containsKey(obj)) {
                            channelContextMapTimeout.put(obj, System.currentTimeMillis() + linkDetectTime);
                        } else {
                            rglog.error("没有找到推送通道更新超时时间");
                        }
                        break;
                }
                throw new NoResponseException();
            } catch (NoResponseException e) {
                throw e;
            } catch (Exception e) {
                // 最终异常处理逻辑
                rglog.error("{}", StringUtil.ExceptionToString(e));
                rglog.error(e.getMessage());
                throw e;
            }
        }
    }

    private class DeameonThread extends Thread {
        @Override
        public void run() {
            this.setName(serviceName);
            RgLogger rglog = BulletinBoard.getLogger(this.getClass().getName());
            rglog.setLogLevel(logLevel);
            while (true) {
                if (!subscribeserver.isAlived()) {
                    subscribeserver.shutdown();
                    subscribeserver.init();
                }
                if (!publishserver.isAlived()) {
                    publishserver.shutdown();
                    publishserver.init();
                }
                Iterator<Map.Entry<Object, Long>> iterator = channelContextMapTimeout.entrySet().iterator();
                long now = System.currentTimeMillis();
                while (iterator.hasNext()) {
                    Map.Entry<Object, Long> entry = iterator.next();
                    if (entry.getValue() < now) {
                        String id = channelContextMapByCtx.get(entry.getKey());
                        rglog.info("订阅者<{}>交互通道长时间未有请求, 从订阅链中删除对应推送通道:{}", id, entry.getKey());
                        channelContextMapByCtx.remove(entry.getKey());
                        channelContextMapTimeout.remove(entry.getKey());
                        List<Object> list = channelContextMapById.get(id);
                        if (list != null && !list.isEmpty()) {
                            list.remove(entry.getKey());
                            rglog.info("订阅者<{}>还剩{}个推送通道", id, list.size());
                        }
                    }
                }
                try {
                    sleep(1000);
                } catch (InterruptedException e) {
                    rglog.error(e.getMessage());
                }
            }
        }
    }

    private class SubscribeMsgPushThread extends Thread {
        @Override
        public void run() {
            this.setName(serviceName);
            RgLogger rglog = BulletinBoard.getLogger(this.getClass().getName());
            rglog.setLogLevel(logLevel);
            while (true) {
                try {
                    rglog.createLogKey();
                    if (cacheMsgFlag) {
                        pushCacheMsg(rglog);
                    } else {
                        String key = "noticeWebSocket:pub:*";
                        RedisUtil.psubscribe(new SubscribeListener(), key);
                    }
                    sleep(200);
                } catch (Exception e) {
                    rglog.error(e.getMessage());
                }
            }
        }

        private void pushCacheMsg(RgLogger rglog) throws Exception {
            synchronized (cacheUserIds) {
                for (String userId : cacheUserIds) {
                    List<Object> ctxs = channelContextMapById.get(userId);
                    if (ctxs == null) {
                        continue;
                    }
                    String popkey = "noticeWebSocket:list:" + userId;
                    List<String> messageList = RedisUtil.getListRange(popkey, 0, -1);
                    if (null != messageList && !messageList.isEmpty()) {
                        rglog.debug("共有{}条消息向订阅者<{}>的{}个通道推送", messageList.size(), userId, ctxs.size());
                        for (String message : messageList) {
                            long timeout = Long.parseLong(message.substring(0, message.indexOf('|')));
                            String responseMsgBase64 = message.substring(message.indexOf('|') + 1);
                            String responeMsg = new String(StringUtil.Base64ToBinary(responseMsgBase64));
                            if (timeout >= System.currentTimeMillis()) {
                                for (Object ctx : ctxs) {
                                    rglog.debug("向订阅者<{}>推送消息:{},推送通道:{}", userId, responeMsg, ctx);
                                    subscribeserver.response(ctx, responeMsg.getBytes(), 0);
                                }
                            } else {
                                rglog.info("订阅者<{}>的缓存消息[{}]已超过缓存时间，不再推送", userId, responeMsg);
                            }
                        }
                        boolean trimFlag = RedisUtil.listTrim(popkey, messageList.size(), -1);
                        rglog.debug("删除已经推送成功的消息列表,trimFlag:{}", trimFlag);
                    }
                }
            }
        }
    }

    private class SubscribeListener extends JedisPubSub {
        @Override
        public void onMessage(String channel, String message) {
            pushSubscribeMsg(channel + "|" + message);
        }

        @Override
        public void onSubscribe(String channel, int subscribedChannels) {
            //订阅了频道会调用
            System.out.printf("subscribe redis channel success, channel %s, subscribedChannels %d%n",
                    channel, subscribedChannels);
        }

        @Override
        public void onUnsubscribe(String channel, int subscribedChannels) {
            //取消订阅 会调用
            System.out.printf("unsubscribe redis channel, channel %s, subscribedChannels %d%n",
                    channel, subscribedChannels);
        }

        @Override
        public void onPMessage(String pattern, String channel, String message) {
            pushSubscribeMsg(channel + "|" + message);
        }

        @Override
        public void onPSubscribe(String pattern, int subscribedChannels) {
            // 初始化按表达式的方式订阅时候的处理
            System.out.printf("psubscribe redis channel success, pattern %s, subscribedChannels %d%n",
                    pattern, subscribedChannels);
        }

        @Override
        public void onPUnsubscribe(String pattern, int subscribedChannels) {
            // 取消按表达式的方式订阅时候的处理
            System.out.printf("punsubscribe redis channel success, pattern %s, subscribedChannels %d%n",
                    pattern, subscribedChannels);
        }

        private void pushSubscribeMsg(String message) {
            if (null != message) {
                String userIdPreString = "noticeWebSocket:pub:";
                String userId = message.substring(userIdPreString.length(), message.indexOf('|'));
                String responseMsgBase64 = message.substring(message.indexOf('|') + 1);
                String responeMsg = new String(StringUtil.Base64ToBinary(responseMsgBase64));
                List<Object> ctxs = channelContextMapById.get(userId);
                if (!ctxs.isEmpty()) {
                    for (Object ctx : ctxs) {
                        rglog.debug("向订阅者<{}>推送消息:{},推送通道:{}", userId, responeMsg, ctx);
                        subscribeserver.response(ctx, responeMsg.getBytes(), 0);
                    }
                } else {
                    rglog.error("订阅者<{}>推送通道异常，不再推送消息:{}", userId, responeMsg);
                }
            }
        }
    }
}
